package com.nokia.position;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.TimeZone;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.nokia.position.PositionProto.Position;
import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;

public class PositionPartitionReducer
		extends
		Reducer<LongWritable, ProtobufWritable<Position>, Text, ProtobufWritable<Position>> {

	private MultipleOutputs<LongWritable, ProtobufWritable<Position>> mos;

	@Override
	protected void reduce(LongWritable key,
			Iterable<ProtobufWritable<Position>> values, Context context)
			throws IOException, InterruptedException {

		for (ProtobufWritable<Position> value : values) {
			/*
			 * setConverter() before get() is required since 'value' object was
			 * created by MR with default ProtobufWritable's default
			 * constructor, as result object does not know its runtime Protobuf
			 * class.
			 */
			value.setConverter(Position.class);
			// context.write(null, new Text(value.get().getName() + "\t"
			// + value.get().getAge()));
			Position position = value.get();
			if (position.hasCoordinatesTimestamp()) {
				Date dateObj = new Date(position.getCoordinatesTimestamp());

				SimpleDateFormat parser = new SimpleDateFormat("yyyy/MM/dd/HH");
				parser.setTimeZone(TimeZone.getTimeZone("UTC"));
				String partitionDate = parser.format(dateObj);
				mos.write(key, value, partitionDate + "/part");
			}
		}
	}

	// public void setup(TaskInputOutputContext<?, ?, LongWritable,
	// ProtobufWritable<Position>> context) {
	// mos = new MultipleOutputs<LongWritable,
	// ProtobufWritable<Position>>(context);
	// }
	public void setup(Context context) {
		mos = new MultipleOutputs(context);
	}

	public void cleanup(Context context) throws IOException {
		try {
			mos.close();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
